#include "StdAfx.h"
#include "SelectModelLib.h"
#include "IAsyncSendQueue.h"

INetworkLib* SelectModelLib::CreateComponent()
{
	return new CSelectModelLib;
}


struct WorkerThreadDesc
{
	SOCKET				m_socket;
	CSelectModelLib*	m_pNetworkLib;	
	WorkerThreadDesc(SOCKET m_socket, CSelectModelLib& networkLib)
		: m_socket(m_socket)
		, m_pNetworkLib(&networkLib)					
	{

	}
};

CSelectModelLib::CSelectModelLib(void)
	: m_pAsyncSendQueue(nullptr)
{
}


CSelectModelLib::~CSelectModelLib(void)
{
	if (m_pAsyncSendQueue)
	{
		m_pAsyncSendQueue->Release();
		m_pAsyncSendQueue = nullptr;
	}
}

bool CSelectModelLib::Initialize( ON_CONNECT_FUNCTOR onConnect , ON_ACCEPT_FUNCTOR onAccept , ON_DISCONNECT_FUNCTOR onDisconnect , ON_RECV_FUNCTOR onRecv )
{
	if (false == Helper::WSAStartup())
	{
		return false;
	}

	m_onConnectFunctor		= onConnect;
	m_onAcceptFunctor		= onAccept;
	m_onDisconnectFunctor	= onDisconnect;
	m_onRecvFunctor			= onRecv;

	m_pAsyncSendQueue = AsyncSendQueueForSelectModel::CreateComponent([&](SOCKET socket){Disconnect(socket);});

	WorkerThreadDesc* pArgment = new WorkerThreadDesc(0, *this);

	UINT threadID = 0;
	HANDLE hThread = (HANDLE)_beginthreadex(0, 0, CSelectModelLib::recvWorkerThread, pArgment, 0, &threadID);
	if (hThread == INVALID_HANDLE_VALUE)
	{
		return false;
	}	

	CloseHandle(hThread);
	return true;
}

bool CSelectModelLib::Connect( const std::tstring& ip, WORD port )
{
	SOCKET connectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (connectSocket == INVALID_SOCKET) 
	{
		return false;
	}

	// Connect to server.
	SOCKADDR_IN sockaddr;	
	INT addressSize = 0;	
	Helper::WSAStringToAddress(ip, port, sockaddr, addressSize);

	int iResult = connect( connectSocket, (SOCKADDR*)&sockaddr, addressSize);
	if (iResult == SOCKET_ERROR) 
	{
		return false;
	}

	m_connectedSocketQueue.push(connectSocket);
	m_onConnectFunctor(connectSocket, 0);

	return true;
}

bool CSelectModelLib::Send( SOCKET socket, IStreamBuffer& stream )
{
	return m_pAsyncSendQueue->Send(socket, stream);
}

bool CSelectModelLib::Send(SOCKET socket)
{
	return m_pAsyncSendQueue->Send(socket, 0);
}

bool CSelectModelLib::Disconnect( SOCKET socket )
{
	shutdown(socket, SD_SEND);
	closesocket(socket);
	return true;
}

bool CSelectModelLib::Listen( const std::tstring& ip, WORD port, int backlog )
{
	SOCKET listenSocket = Helper::WSAListen(ip, port, backlog);
	if (listenSocket == INVALID_SOCKET)
	{
		return false;
	}

	WorkerThreadDesc* pArgment = new WorkerThreadDesc(listenSocket, *this);

	UINT threadID = 0;
	HANDLE hThread = (HANDLE)_beginthreadex(0, 0, CSelectModelLib::acceptWorkerThread, pArgment, 0, &threadID);
	if (hThread == INVALID_HANDLE_VALUE)
	{
		Disconnect(listenSocket);
		return false;
	}	

	CloseHandle(hThread);
	return true;
}

void CSelectModelLib::Release()
{
	Helper::WSACleanup();
	delete this;
}


UINT __stdcall CSelectModelLib::recvWorkerThread( void* pArguments )
{
	WorkerThreadDesc* pWorkerThreadDesc = (WorkerThreadDesc*)pArguments;
	typedef std::list<SOCKET>	CONNECTED_SESSIONS;
	CONNECTED_SESSIONS connectedSessions;
	while(1)
	{		
		while(1)
		{
			SOCKET connectedSocket;
			if (pWorkerThreadDesc->m_pNetworkLib->m_connectedSocketQueue.try_pop(connectedSocket) == false)
			{
				break;
			}

			connectedSessions.push_back(connectedSocket);
		}

		std::list<CONNECTED_SESSIONS::iterator> closedSessionIterators;

		fd_set ReadFDs, WriteFDs, ExceptFDs;
		FD_ZERO(&ReadFDs);
		FD_ZERO(&WriteFDs);
		FD_ZERO(&ExceptFDs);

		for each(SOCKET socket in connectedSessions)
		{
			FD_SET(socket, &ReadFDs);
			FD_SET(socket, &WriteFDs);
			FD_SET(socket, &ExceptFDs);
		}

		timeval tv;
		tv.tv_sec = 0;
		tv.tv_usec = 10;

		if (select(0, &ReadFDs, &WriteFDs, &ExceptFDs, 0) > 0) 
		{
			
			auto it = connectedSessions.begin();
			auto it_end = connectedSessions.end();
			for(; it != it_end; ++it)
			{
				const char* pcErrorType = 0;

				// See if this socket's flag is set in any of the FD
				// sets.
				SOCKET socket = (*it);
				if (FD_ISSET(socket, &ExceptFDs)) 
				{
					FD_CLR(socket, &ExceptFDs);
				}
				else 
				{
					if (FD_ISSET(socket, &ReadFDs)) 
					{
						int iResult = 0;

						char	tempBuf[512];
						iResult = recv(socket, tempBuf, sizeof(tempBuf), 0);
						if ( iResult > 0 )
						{
							IStreamBuffer* pStreamBuffer = StreamBuffer::CreateComponent();
							pStreamBuffer->Write(tempBuf, iResult);
							pWorkerThreadDesc->m_pNetworkLib->m_onRecvFunctor(socket, pStreamBuffer);
						}
						else if ( iResult == 0 )
						{
							closedSessionIterators.push_back(it);
							break;
						}
						else
						{
							if (WSAEWOULDBLOCK != WSAGetLastError())
							{
								closedSessionIterators.push_back(it);
								break;
							}
						}

						FD_CLR(socket, &ReadFDs);
					}	
					if (FD_ISSET(socket, &WriteFDs)) 
					{	
						pWorkerThreadDesc->m_pNetworkLib->Send(socket);

						FD_CLR(socket, &WriteFDs);
					}
				}
			}	
			for each(auto item in closedSessionIterators)
			{
				pWorkerThreadDesc->m_pNetworkLib->Disconnect(*item);
				pWorkerThreadDesc->m_pNetworkLib->m_onDisconnectFunctor(*item);

				connectedSessions.erase(item);
			}
		}	

	}

	delete pWorkerThreadDesc;

	return 0;
}


UINT __stdcall CSelectModelLib::acceptWorkerThread( void* pArguments )
{
	WorkerThreadDesc* pWorkerThreadDesc = (WorkerThreadDesc*)pArguments;
	
	while(1)
	{
		
		fd_set ReadFDs, WriteFDs, ExceptFDs;
		FD_ZERO(&ReadFDs);
		FD_ZERO(&WriteFDs);
		FD_ZERO(&ExceptFDs);

		FD_SET(pWorkerThreadDesc->m_socket, &ReadFDs);
		FD_SET(pWorkerThreadDesc->m_socket, &WriteFDs);
		FD_SET(pWorkerThreadDesc->m_socket, &ExceptFDs);
		
		if (select(0, &ReadFDs, &WriteFDs, &ExceptFDs, 0) > 0) 
		{
			//// Something happened on one of the sockets.
			// Was it the listener socket?...
			if (FD_ISSET(pWorkerThreadDesc->m_socket, &ReadFDs)) 
			{
				SOCKET clientSocket = accept(pWorkerThreadDesc->m_socket, NULL, NULL);

				if (clientSocket == INVALID_SOCKET) 
				{
					continue;
				}

				u_long nNoBlock = 1;
				ioctlsocket(clientSocket, FIONBIO, &nNoBlock);
				pWorkerThreadDesc->m_pNetworkLib->m_connectedSocketQueue.push(clientSocket);

				pWorkerThreadDesc->m_pNetworkLib->m_onAcceptFunctor(clientSocket);
				
			}
			else if (FD_ISSET(pWorkerThreadDesc->m_socket, &ExceptFDs)) 
			{
				break;
			}
			
		}		
	}

	pWorkerThreadDesc->m_pNetworkLib->Disconnect(pWorkerThreadDesc->m_socket);
	delete pWorkerThreadDesc;

	return 0;
}


